package com.dtwave.cfstreaming.process;

import com.alibaba.fastjson.JSONObject;
import com.dtwave.cfstreaming.constant.SQLFileConstants;
import com.dtwave.cfstreaming.update.StreamDimUpdate;
import com.dtwave.cfstreaming.utils.FileUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class StreamDataExecutor {
    public static void excuteFoResult(StreamTableEnvironment tableEnv, ParameterTool params, SingleOutputStreamOperator<JSONObject> flatMapStream) {
        //注册流表
        TableRegister.streamTableFromKafka(tableEnv,flatMapStream);
        //注册维度（可以参与到后面的sql处理中需要有时间字段作为时态）
        TableRegister.dimTableFromHbase(tableEnv,params);
        //维表更新
        StreamDimUpdate.updateTableToHbase(flatMapStream);
        //执行sql文件生成flinkSql中的表,可以使用上面处理得到的流表参与sql处理（middleTab）
        String sql = FileUtils.getSqlStream(StreamDataExecutor.class, SQLFileConstants.TEST_SQL_FILE);
        tableEnv.executeSql(sql);
    }
}
